package sgt.project;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.spout.FirstPollOffsetStrategy;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;
import sgt.test.CountBolt;
import sgt.test.InitBolt;

public class MakeUpTopo {
    public static void main(String[] args) throws Exception{
        //创建TopologyBuilder
        TopologyBuilder topologyBuilder = new TopologyBuilder();

        KafkaSpoutConfig.Builder<String, String> kafkaSpoutConfigBuilder;
        //kafka连接信息
        String bootstrapServers = "niit-master:9092";

        String topic = "mytopic";

        kafkaSpoutConfigBuilder = new KafkaSpoutConfig.Builder(
                bootstrapServers,
                topic);

        //使用kafkaSpoutConfigBuilder构造器构造kafkaSpoutConfig,并配置相应属性
        KafkaSpoutConfig<String, String> kafkaSpoutConfig = kafkaSpoutConfigBuilder
                /**
                 * 设置groupId
                 */
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, topic.toLowerCase() + "_storm_group")

                /**
                 * 设置session超时时间,该值应介于
                 * [group.min.session.timeout.ms, group.max.session.timeout.ms] [6000,300000]
                 * 默认值:10000
                 */
                .setProp(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "100000")

                /**
                 * 设置拉取最大容量
                 */
                .setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "1048576")

                /**
                 * 设置控制客户端等待请求响应的最大时间量
                 * 默认值:30000
                 */
                .setProp(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "300000")

                /**
                 * 设置心跳到消费者协调器之间的预期时间。
                 * 心跳用于确保消费者的会话保持活动并且当新消费者加入或离开组时促进重新平衡
                 * 默认值:	3000        (一般设置低于session.timeout.ms的三分之一)
                 */
                .setProp(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "30000")

                /**
                 * 设置offset提交时间15s  默认30s
                 */
                .setOffsetCommitPeriodMs(15000)

//                .setProp(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest")
                .setFirstPollOffsetStrategy(FirstPollOffsetStrategy.LATEST)
                /**
                 * 设置配置消息的key/value的反序列化类
                 */
                .setProp("key.deserializer", StringDeserializer.class)
                .setProp("value.deserializer",StringDeserializer.class)

                /**
                 * 构造kafkaSpoutConfig
                 */
                .build();


        //setSpout
        topologyBuilder.setSpout("kafkaSpout", new KafkaSpout(kafkaSpoutConfig));
        topologyBuilder.setBolt("MakeUpInitBolt",new MakeUpInitBolt()).localOrShuffleGrouping("kafkaSpout");
        topologyBuilder.setBolt("MakeUpCountBolt",new MakeUpCountBolt()).localOrShuffleGrouping("MakeUpInitBolt");
//        topologyBuilder.setBolt("MaxTenCartBolt",new MaxTenCartBolt()).localOrShuffleGrouping("MakeUpCountBolt");
//        topologyBuilder.setBolt("InitBolt", new InitBolt()).localOrShuffleGrouping("kafkaSpout");
//        topologyBuilder.setBolt("CountBolt",new CountBolt()).localOrShuffleGrouping("InitBolt");

        Config config = new Config();
        /**
         * 设置supervisor和worker之间的通信超时时间.
         * 超过这个时间supervisor会重启worker  (秒)
         */
        config.put("supervisor.worker.timeout.secs", 600000);
        /**
         * 设置storm和zookeeper之间的超时时间.
         */
        config.put("storm.zookeeper.session.timeout", 1200000000);
        /**
         * 设置debug模式 日志输出更全
         * 只能在本地LocalCluster模式下启用
         */
        config.setDebug(true);
//        StormSubmitter.submitTopology("NewKafKaTopo", config, topologyBuilder.createTopology());
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("NewKafKaTopo", config, topologyBuilder.createTopology());
//        Utils.sleep(Long.MAX_VALUE);
//        localCluster.shutdown();


    }
}
